package com.demo.kafka;

import java.util.Properties;
import java.util.Scanner;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class KafkaProduce {
	
	private static Properties props;
	
	public static void init() {
		props = new Properties();
		props.put("bootstrap.servers", "192.168.6.131:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
	}

	public static void main(String[] args) {
		init();
		KafkaProducer<String, String> producer = new KafkaProducer<>(props);
		
		Scanner scanner = new Scanner(System.in);
		
		while (true) {
            System.out.println("请输入要发送的消息：");
            String value = scanner.nextLine();
            if (value.equals("exit")) {
            	scanner.close();
            	producer.close();
                break;
            }

            producer.send(new ProducerRecord<String, String>("kafka-topics", value));
        }
	}
}
